package net.butfly.bus.server.filter.impl;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import net.butfly.bus.Request;
import net.butfly.bus.Response;
import net.butfly.bus.context.Constants;
import net.butfly.bus.exception.BusRuntimeException;
import net.butfly.bus.filter.FilterBase;
import net.butfly.bus.util.LogUtils;

public class ThreadController extends FilterBase {
	private static LogUtils logger = LogUtils.getLog();
	private final static String key_throughputName = "throughput";
	private final static String key_snapshotEnabled = "snapshotEnabled";
	private final static String snapshotIntervalTime = "snapshotIntervalTime";
	private List<Object[]> curResList = new ArrayList<Object[]>();
	private long snapshotLastTime = 0;

	@Override
	public void execute(Request request, Response response) {
		int counter = getInitParamAsInt(key_throughputName);
		boolean snapshotEnabled = getInitParamAsBoolean(key_snapshotEnabled);
		int snapshotStepTime = getInitParamAsInt(snapshotIntervalTime);

		if (counter > 0) {
			if (curResList.size() > 0) {
				Iterator<Object[]> ite = curResList.iterator();
				while (ite.hasNext()) {
					if (((Response) ite.next()[1]).isFinished()) {
						ite.remove();
					}
				}
			}
			synchronized (curResList) {
				if (curResList.size() < counter) {
					curResList.add(new Object[] { request, response });
				} else {
					logger.error(LogUtils.getInstance().getFlowNo() + " thread limit (" + counter + ") overflow");
					throw new BusRuntimeException(Constants.SystemError.THREAD_COUNTER_OVERFLOW,
							"System busy please retry after some seconds.");
				}
			}
			if (snapshotEnabled) {
				if (snapshotLastTime == 0) {
					snapshotLastTime = System.currentTimeMillis();
				} else if (System.currentTimeMillis() - snapshotLastTime > snapshotStepTime) {
					logger.info(LogUtils.getInstance().getFlowNo()
							+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date())
							+ " Bus snapshot - current thread count: " + curResList.size() + ", current transaction: ");
					for (int i = 0; i < curResList.size(); i++) {
						logger.info(LogUtils.getInstance().getFlowNo() + " bus snapshot - current transaction[" + i
								+ "]:" + ((Request) curResList.get(i)[0]).getTx() + "["
								+ ((Request) curResList.get(i)[0]).getVersion() + "]");
					}
					snapshotLastTime = System.currentTimeMillis();
				}
			}
		} else {
			throw new BusRuntimeException(Constants.UserError.CONFIG_ERROR, "thread limit is ZERO.");
		}
	}
}
